package com.examples.mysql2neo4j;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.neo4j.driver.AuthTokens;
import org.neo4j.driver.Driver;
import org.neo4j.driver.GraphDatabase;
import org.neo4j.driver.Session;

import java.sql.*;

public class TwitterDemo_tweet_retweetship_relationship {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<JSONObject> userDataStream = env
                .addSource(new RichSourceFunction<JSONObject>() {
                    private static final long serialVersionUID = 3334652233018091675L;
                    private Connection connect = null;
                    private PreparedStatement ps = null;
                    private boolean running=false;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        running =true;
                        Class.forName("com.mysql.jdbc.Driver");
                        connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.0.117:3300/twitter_spider?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");
                        ps = (PreparedStatement) connect
                                .prepareStatement("select * from tw_tweet_retweetship");
                    }

                    @Override
                    public void run(SourceContext<JSONObject> sourceContext) throws Exception {
                        while(running){
                            ResultSet resultSet = ps.executeQuery();
                            while (resultSet.next()) {
                                JSONObject jsonObject = resultSetToJsonObject(resultSet);
                                sourceContext.collect(jsonObject);
                            }
                        }
                    }

                    public JSONObject resultSetToJsonObject(ResultSet rs) throws SQLException, JSONException {
                        JSONObject jsonObj = new JSONObject();
                        ResultSetMetaData metaData = rs.getMetaData();
                        int columnCount = metaData.getColumnCount();
                        for (int i = 1; i <= columnCount; i++) {
                            String columnName =metaData.getColumnLabel(i);
                            String value = rs.getString(columnName);
                            jsonObj.put(columnName, value);
                        }
                        return jsonObj;
                    }

                    @Override
                    public void cancel() {
                        running =false;
                        try {
                            super.close();
                            if (connect != null) {
                                connect.close();
                            }
                            if (ps != null) {
                                ps.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).name("抽取 tw_user_relationship 数据").setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, String>> tw_user_relationship_map = userDataStream.map(new MapFunction<JSONObject, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(JSONObject jsonObject) throws Exception {
                return new Tuple2<>(jsonObject.getString("tweet_id"), jsonObject.getString("src_tweet_id"));
            }
        }).name("转换为 tw_user_relationship map").setParallelism(10);
        tw_user_relationship_map.addSink(new SinkFunction<Tuple2<String, String>>() {
            @Override
            public void invoke(Tuple2<String, String> value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                String uri = "bolt://192.168.0.117:7687";
                String user = "neo4j";
                String password = "12345678";
                try (Driver driver = GraphDatabase.driver(uri, AuthTokens.basic(user, password))) {
                    try (Session session = driver.session()) {
                        String s = "MATCH (n1:TwUserTweet{id: '"+value.f0+"'}) , (n2:TwUserTweet{id: '"+value.f1+"'})  WITH n1, n2 WHERE ID(n1) IS NOT NULL AND ID(n2) IS NOT NULL OPTIONAL MATCH (n1)-[r:RETWEETSHIP]->(n2) FOREACH(ignoreMe IN CASE WHEN r IS NULL THEN [1] ELSE [] END | CREATE (n1)-[:RETWEETSHIP]->(n2) ) RETURN n1, n2;";
                        session.run(s);
                        System.out.println("Run SQL to Neo4j: "+s);
                    }
                } catch (Exception e) {
                    System.err.println("Error connecting to Neo4j: " + e.getMessage());
                }
            }
        }).name("在图谱中创建关系").setParallelism(1);
        env.execute("TwitterDemo_user_tweet");


    }
}
